﻿#ifndef JCONCURRENT_ITERATEKERNEL_H
#define JCONCURRENT_ITERATEKERNEL_H

#include "jconcurrent_global.h"
#include "jatomic.h"
#include "jconcurrentmedian.h"
#include "jconcurrentthreadengine.h"
#include <vector>
#include <iterator>

namespace JConcurrent {

/*
    The BlockSizeManager class manages how many iterations a thread should
    reserve and process at a time. This is done by measuring the time spent
    in the user code versus the control part code, and then increasing
    the block size if the ratio between them is to small. The block size
    management is done on the basis of the median of several timing measuremens,
    and it is done induvidualy for each thread.
*/
class JCONCURRENT_EXPORT BlockSizeManager
{
public:
    BlockSizeManager(int iterationCount);
    void timeBeforeUser();
    void timeAfterUser();
    int blockSize();

private:
    inline bool blockSizeMaxed()
    {
        return (m_blockSize >= maxBlockSize);
    }

private:
    const int maxBlockSize;
    jint64 beforeUser;
    jint64 afterUser;
#if defined(_MSC_VER)
#pragma warning(push)
#pragma warning ( disable: 4251 )
#endif
    Median<double> controlPartElapsed;
    Median<double> userPartElapsed;
#if defined(_MSC_VER)
#pragma warning(pop)
#endif
    int m_blockSize;
};

template <typename T>
class ResultReporter
{
public:
    ResultReporter(ThreadEngine<T> *_threadEngine)
        : threadEngine(_threadEngine)
    {

    }

    void reserveSpace(int resultCount)
    {
        currentResultCount = resultCount;
        vector.resize(jMax<int>(resultCount, vector.size()));
    }

    void reportResults(int begin)
    {
        const int useVectorThreshold = 4; // Tunable parameter.
        if (currentResultCount > useVectorThreshold) {
            vector.resize(currentResultCount);
            threadEngine->reportResults(vector, begin);
        } else {
            for (int i = 0; i < currentResultCount; ++i)
                threadEngine->reportResult(&vector.at(i), begin + i);
        }
    }

    inline T *getPointer()
    {
        return &vector[0];
    }

    int currentResultCount;
    ThreadEngine<T> *threadEngine;
    jvector<T> vector;
};

template <>
class ResultReporter<void>
{
public:
    inline ResultReporter(ThreadEngine<void> *) { }
    inline void reserveSpace(int) { }
    inline void reportResults(int) { }
    inline void *getPointer() { return 0; }
};

inline bool selectIteration(std::bidirectional_iterator_tag)
{
    return false; // while
}

inline bool selectIteration(std::forward_iterator_tag)
{
    return false; // while
}

inline bool selectIteration(std::random_access_iterator_tag)
{
    return true; // for
}

template <typename Iterator, typename T>
class IterateKernel : public ThreadEngine<T>
{
public:
    typedef T ResultType;

    IterateKernel(Iterator _begin, Iterator _end)
        : begin(_begin), end(_end), current(_begin), currentIndex(0)
        , forIteration(selectIteration(typename std::iterator_traits<Iterator>::iterator_category()))
        , progressReportingEnabled(true)
    {
        iterationCount =  forIteration ? std::distance(_begin, _end) : 0;
    }

    virtual ~IterateKernel() { }

    virtual bool runIteration(Iterator it, int index , T *result)
        { J_UNUSED(it); J_UNUSED(index); J_UNUSED(result); return false; }
    virtual bool runIterations(Iterator _begin, int beginIndex, int endIndex, T *results)
        { J_UNUSED(_begin); J_UNUSED(beginIndex); J_UNUSED(endIndex); J_UNUSED(results); return false; }

    void start()
    {
        progressReportingEnabled = this->isProgressReportingEnabled();
        if (progressReportingEnabled && iterationCount > 0) {
            this->setProgressRange(0, iterationCount);
        }
    }

    bool shouldStartThread()
    {
        if (forIteration) {
            return (currentIndex < iterationCount) && !this->shouldThrottleThread();
        } else { // whileIteration
            return (iteratorThreads == 0);
        }
    }

    ThreadFunctionResult threadFunction()
    {
        if (forIteration) {
            return this->forThreadFunction();
        } else { // whileIteration
            return this->whileThreadFunction();
        }
    }

    ThreadFunctionResult forThreadFunction()
    {
        BlockSizeManager blockSizeManager(iterationCount);
        ResultReporter<T> resultReporter(this);

        for(;;) {
            if (this->isCanceled()) {
                break;
            }

            const int currentBlockSize = blockSizeManager.blockSize();

            if (currentIndex >= iterationCount) {
                break;
            }

            // Atomically reserve a block of iterationCount for this thread.
            const int beginIndex = currentIndex.fetchAndAddRelease(currentBlockSize);
            const int endIndex = jMin(beginIndex + currentBlockSize, iterationCount);

            if (beginIndex >= endIndex) {
                break;  // No more work
            }

            this->waitForResume(); // (only waits if the jfuture is paused.)

            if (shouldStartThread()) {
                this->startThread();
            }

            const int finalBlockSize = endIndex - beginIndex; // block size adjusted for possible end-of-range
            resultReporter.reserveSpace(finalBlockSize);

            // Call user code with the current iteration range.
            blockSizeManager.timeBeforeUser();
            const bool resultsAvailable = this->runIterations(begin, beginIndex, endIndex, resultReporter.getPointer());
            blockSizeManager.timeAfterUser();

            if (resultsAvailable)
                resultReporter.reportResults(beginIndex);

            // Report progress if progress reporting enabled.
            if (progressReportingEnabled) {
                completed.fetchAndAddAcquire(finalBlockSize);
                this->setProgressValue(this->completed);
            }

            if (this->shouldThrottleThread())
                return ThrottleThread;
        }
        return ThreadFinished;
    }

    ThreadFunctionResult whileThreadFunction()
    {
        if (iteratorThreads.testAndSetAcquire(0, 1) == false) {
            return ThreadFinished;
        }

        ResultReporter<T> resultReporter(this);
        resultReporter.reserveSpace(1);

        while (current != end) {
            // The following two lines breaks support for input iterators according to
            // the sgi docs: dereferencing prev after calling ++current is not allowed
            // on input iterators. (prev is dereferenced inside user.runIteration())
            Iterator prev = current;
            ++current;
            int index = currentIndex.fetchAndAddRelaxed(1);
            iteratorThreads.testAndSetRelease(1, 0);

            this->waitForResume(); // (only waits if the qfuture is paused.)

            if (shouldStartThread()) {
                this->startThread();
            }

            const bool resultAavailable = this->runIteration(prev, index, resultReporter.getPointer());
            if (resultAavailable) {
                resultReporter.reportResults(index);
            }

            if (this->shouldThrottleThread()) {
                return ThrottleThread;
            }

            if (iteratorThreads.testAndSetAcquire(0, 1) == false) {
                return ThreadFinished;
            }
        }

        return ThreadFinished;
    }

public:
    const Iterator begin;
    const Iterator end;
    Iterator current;
    JAtomicInt currentIndex;
    bool forIteration;
    JAtomicInt iteratorThreads;
    int iterationCount;

    bool progressReportingEnabled;
    JAtomicInt completed;
};

} // namespace JConcurrent

#endif  // JCONCURRENT_ITERATEKERNEL_H
